package woaini.wind.bot.chat.impl;

import cn.hutool.cache.impl.TimedCache;
import cn.hutool.core.collection.BoundedPriorityQueue;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import com.coze.openapi.client.chat.CreateChatReq;
import com.coze.openapi.client.chat.model.*;
import com.coze.openapi.client.connversations.message.model.Message;
import com.coze.openapi.service.auth.JWTOAuth;
import com.coze.openapi.service.auth.JWTOAuthClient;
import com.coze.openapi.service.service.CozeAPI;
import io.reactivex.Flowable;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.stereotype.Component;
import woaini.wind.bot.chat.AbstractChatter;
import woaini.wind.bot.chat.config.CozeChatterConfig;
import woaini.wind.bot.chat.dto.ChatMessage;
import woaini.wind.bot.chat.dto.UserInputDTO;
import woaini.wind.bot.core.exception.BotAssertTool;
import woaini.wind.bot.core.exception.enums.BotExceptionType;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 扣子聊天api接口
 *
 * @see woaini.wind.bot.chat.impl.CozeChatter
 * @author yefeng {@date 2025-03-18 14:13:07}
 */
@Slf4j
@Component
@EnableConfigurationProperties(CozeChatterConfig.class)
public class CozeChatter extends AbstractChatter {

  private final CozeAPI cozeAPI;
  public static final Set<Character> CHAT_END_PUNCTUATION =
      CollUtil.newHashSet('.', '!', '?', '。', '！', '？', '~');

  TimedCache<String, String> CHAT_CONVERSATION_MAP = new TimedCache<>(1000 * 60 * 3);

  private static final ThreadFactory FACTORY =
      Thread.ofVirtual()
          .name("virtual-thread-bot-2-", 1)
          .uncaughtExceptionHandler(
              (t, e) ->
                  log.error("虚拟线程Id:{}, 虚拟线程名:{}, 依附的平台线程:{}", t.threadId(), t.getName(), t, e))
          .factory();
  private static final ExecutorService executorService =
      Executors.newThreadPerTaskExecutor(FACTORY);

  public CozeChatter(CozeChatterConfig config, CozeAPI cozeAPI) {
    super(config);
    this.cozeAPI = cozeAPI;
  }

  @Override
  public String handlerChat(String userId, String userName, List<ChatMessage> messages) {

    CozeChatterConfig baseConfig = (CozeChatterConfig) this.chatterBaseConfig;

    CreateChatReq req = this.getCreateChatReq(userId, userName, messages, baseConfig);

    try {
      ChatPoll chat = cozeAPI.chat().createAndPoll(req);
      return chat.getMessages().get(0).getContent();
    } catch (Exception e) {
      log.error("coze chat error", e);
      BotAssertTool.isActualTrue(
          true, BotExceptionType.BOT_RPC_PARAMS, "异常:\r\n{}", e.getMessage());
    }
    return "";
  }

  private CreateChatReq getCreateChatReq(
      String userId, String userName, List<ChatMessage> messages, CozeChatterConfig baseConfig) {
    List<Message> sendMessages = new ArrayList<>(messages.size());
    // 消息转换
    for (ChatMessage message : messages) {
      String role = message.getRole();
      UserInputDTO userInputDTO = new UserInputDTO();
      String content = message.getContent();
      userInputDTO.setUserId(userId);
      userInputDTO.setUserName(userName);
      userInputDTO.setContent(content);
      if (role.equals("user")) {
        Message sendMessage = Message.buildUserQuestionText(JSONObject.toJSONString(userInputDTO));
        sendMessages.add(sendMessage);
      } else if (role.equals("assistant")) {
        Message sendMessage = Message.buildAssistantAnswer(JSONObject.toJSONString(userInputDTO));
        sendMessages.add(sendMessage);
      }
    }
    CreateChatReq req =
        CreateChatReq.builder()
            .botID(baseConfig.getBotId())
            .userID(userId)
            .messages(sendMessages)
            .build();
    return req;
  }

  @Override
  public void streamHandlerChat(
      BoundedPriorityQueue<ChatMessage> cacheMessageQueue,
      SubmissionPublisher<String> publisher,
      String userId,
      String userName,
      List<ChatMessage> messages) {
    CozeChatterConfig baseConfig = (CozeChatterConfig) this.chatterBaseConfig;

    CreateChatReq req = this.getCreateChatReq(userId, userName, messages, baseConfig);

    Flowable<ChatEvent> resp = cozeAPI.chat().stream(req);
    executorService.submit(
        () -> {
          this.handlerChatEvent(cacheMessageQueue, resp, publisher);
        });
  }

  /**
   * 处理程序聊天事件
   *
   * @param resp 回复
   * @param publisher 发行人
   * @author yefeng {@date 2025-06-22 17:53:13}
   * @since 1.0
   */
  private void handlerChatEvent(
      BoundedPriorityQueue<ChatMessage> cacheMessageQueue,
      Flowable<ChatEvent> resp,
      SubmissionPublisher<String> publisher) {

    AtomicReference<ChatEvent> pluginEventRef = new AtomicReference<>(null);
    AtomicReference<String> conversationRef = new AtomicReference<>(null);

    StringBuilder cacheStringBuilder = new StringBuilder();

    AtomicBoolean eventOver = new AtomicBoolean(false);
    AtomicBoolean messageOver = new AtomicBoolean(false);

    resp.blockingSubscribe(
        event -> {
          log.info(
              "coze chat event:{}-{}",
              event.getEvent().getValue(),
              ObjectUtil.isNotEmpty(event.getMessage()) ? event.getMessage().getContent() : "");
          ChatEventType chatEventType = event.getEvent();
          // 正常消息流
          if (chatEventType == ChatEventType.CONVERSATION_MESSAGE_DELTA) {
            String content = event.getMessage().getContent();
            this.appendContentAndSend(publisher, messageOver, cacheStringBuilder, content);
          }
          // 工具调用
          if (chatEventType == ChatEventType.CONVERSATION_CHAT_REQUIRES_ACTION) {
            pluginEventRef.set(event);
            conversationRef.set(event.getChat().getConversationID());
          }
          if (chatEventType == ChatEventType.CONVERSATION_MESSAGE_COMPLETED) {

            String content = event.getMessage().getContent();
            if (!content.startsWith("{\"") && cacheMessageQueue != null) {
              log.info("存储历史消息:{}", content);
              // 历史记录存储
              cacheMessageQueue.add(ChatMessage.ofSystem(content));
            }
          }
          if (chatEventType == ChatEventType.CONVERSATION_CHAT_COMPLETED
              || chatEventType == ChatEventType.DONE) {
            eventOver.set(true);
          }
        });

    if (pluginEventRef.get() != null) {
      // 工具调用处理
      List<ToolOutput> toolOutputs = new ArrayList<>();

      ChatEvent pluginEvent = pluginEventRef.get();
      for (ChatToolCall callInfo :
          pluginEvent.getChat().getRequiredAction().getSubmitToolOutputs().getToolCalls()) {
        String callID = callInfo.getID();
        String functionName = callInfo.getFunction().getName();
        String argsJson = callInfo.getFunction().getArguments();

        log.info("工具调用:{}-{}", functionName, argsJson);
        toolOutputs.add(ToolOutput.of(callID, "未查询到数据"));
      }
    }

    while (true) {
      // 消息都结束的情况下
      if (eventOver.get() && messageOver.get()) {
        if (StrUtil.isNotBlank(cacheStringBuilder.toString())) {
          publisher.submit(cacheStringBuilder.toString());
        }
        // 处理完毕，主动关闭发送
        publisher.close();
        break;
      }
    }
    log.info("对话结束");
  }

  /**
   * 如果满足一句话的条件，则追加内容并发送
   *
   * @param publisher 发行人
   * @param sb 某人
   * @param content 内容
   * @author yefeng {@date 2025-06-24 14:14:55}
   * @since 1.0
   */
  private void appendContentAndSend(
      SubmissionPublisher<String> publisher,
      AtomicBoolean messageOver,
      StringBuilder sb,
      String content) {
    messageOver.set(false);
    // 判断是否满足一句话的条件
    for (int i = 0; i < content.length(); i++) {
      char currentChar = content.charAt(i);
      sb.append(currentChar);
      if (CHAT_END_PUNCTUATION.contains(currentChar)) {
        if (!publisher.isClosed()) {
          log.info("发送消息:{}", sb.toString());
          publisher.submit(sb.toString());
        }
        sb.setLength(0);
      }
    }
    messageOver.set(true);
  }

  @Override
  public String useInfo() {
    return "";
  }
}
